package exp1;


import java.io.IOException;
import java.util.StringTokenizer;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import example.WordCount;
import example.WordCount.IntSumReducer;
import example.WordCount.TokenizerMapper;

public class Lab1 {
	//Mapper<keyin ,valuein ,keyout, valueout>
	public static class Map extends Mapper<Object,Text,Text,Text>{
		private static Text text = new Text();
		public void map(Object key,Text value,Context context)
				throws  IOException, InterruptedException{
			text = value;
			context.write(value, new Text(""));//value 值记为空
		}
	}
	
	//Reducer<keyin, valuein, keyout, valueout>
	public static class Reduce extends Reducer<Text,Text,Text,Text> {
		private static Text text = new Text();
		
		//这一不，reduce 每次调用，一个key ，一组alue
		public void reduce(Text key, Iterable<Text> values, 
                    Context context
                    ) throws IOException, InterruptedException {
			
			context.write(key, new Text(""));
		}
	}
	public static void main(String[] args) throws Exception{
		  args = new String[] {"hdfs://localhost:9000/ex1/input1","hdfs://localhost:9000/ex1/output1"}; 
		  Configuration conf = new Configuration();
		  conf.set("fs.defaultFS",  "hdfs://localhost:9000");
		  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		  if (otherArgs.length != 2) {
			  System.err.println("Usage: wordcount <in> <out>");
			  System.exit(2);
		  }
		  Path path =new Path(args[1]);
	      //加载配置文件 
	      FileSystem fileSystem = path.getFileSystem(conf); 
	      //输出目录若存在则删除 
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
		  Job job = new Job(conf, "lab1");
		  job.setJarByClass(Lab1.class);
		  job.setMapperClass(Map.class);
		  job.setCombinerClass(Reduce.class);
		  job.setReducerClass(Reduce.class);
		  job.setOutputKeyClass(Text.class);
		  job.setOutputValueClass(Text.class);
		  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		  System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
